package com.tiger.cloud.common.redis.queue;

import org.apache.log4j.Logger;

import com.google.common.util.concurrent.AbstractExecutionThreadService;

/**
 * 这个类功能和RedisMessageConsumer类一样，这里采用的是Guava的Service框架来简化服务的生命周期管理。
 */
public class MessageConsumer extends AbstractExecutionThreadService {

	private static Logger log = Logger.getLogger(MessageConsumer.class);

	private volatile boolean RUNNING = true;
	private RedisQueue redisQueue;
	private MessageListener listener;

	public MessageConsumer(RedisQueue redisQueue) {
		this.redisQueue = redisQueue;
	}

	public MessageConsumer(RedisQueue redisQueue, MessageListener listener) {
		this.redisQueue = redisQueue;
		this.listener = listener;
	}

	public void setListener(MessageListener listener) {
		this.listener = listener;
	}

	@Override
	protected void run() throws Exception {
		while (RUNNING) {
			try {
				String message = redisQueue.askForMessage(2); // brpoplpush 2s
				if (message != null && message.length() > 0) {
					listener.onMessage(message);
					redisQueue.onCompleted(message);
				} else {
					Thread.sleep(1000);
				}
			} catch (Exception e) {
				log.error(e.getMessage(), e);
			}
		}
	}

	@Override
	protected void triggerShutdown() {
		RUNNING = false;
	}

}
